//! Tokio网络层具体实现
//! 
//! 实现core模块定义的抽象trait，提供Tokio特定的网络和运行时能力
//! 这些实现会传递给协议层使用

use std::sync::{Arc, atomic::{AtomicU64, Ordering}};
use tokio::net::TcpStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::core::network::{NetworkStream, LightweightRuntime, TaskHandle, RuntimeAbstraction, RuntimeStats};
use std::future::Future;
use std::pin::Pin;
use std::time::Duration;

/// Tokio网络流适配器
/// 
/// 将Tokio的异步TcpStream适配为同步NetworkStream接口
/// 使得协议层可以同步使用，而网络层保持异步
#[derive(Debug)]
pub struct TokioNetworkStream {
    /// 连接地址信息
    remote_addr: Option<String>,
    /// 连接状态
    connected: bool,
    /// 内部缓冲区（模拟同步读写）
    read_buffer: Vec<u8>,
    /// 写入缓冲区
    write_buffer: Vec<u8>,
}

impl TokioNetworkStream {
    /// 从Tokio TcpStream创建适配器
    /// 
    /// 注意：这是一个简化实现，实际使用中需要更复杂的适配逻辑
    /// 因为NetworkStream trait要求同步接口，而TcpStream是异步的
    pub async fn from_tokio_stream(stream: TcpStream) -> std::io::Result<Self> {
        let remote_addr = stream.peer_addr()
            .map(|addr| addr.to_string())
            .ok();
        
        // 注意：这里我们不能直接存储TcpStream，因为NetworkStream需要同步接口
        // 实际实现中，这需要更复杂的异步桥接机制
        
        Ok(Self {
            remote_addr,
            connected: true,
            read_buffer: Vec::new(),
            write_buffer: Vec::new(),
        })
    }
    
    /// 创建用于测试的模拟网络流
    pub fn mock_for_testing(remote_addr: String) -> Self {
        Self {
            remote_addr: Some(remote_addr),
            connected: true,
            read_buffer: Vec::new(),
            write_buffer: Vec::new(),
        }
    }
}

impl NetworkStream for TokioNetworkStream {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        // 简化实现：从内部缓冲区读取
        if self.read_buffer.is_empty() {
            return Ok(0);  // 没有数据
        }
        
        let to_copy = std::cmp::min(buf.len(), self.read_buffer.len());
        buf[..to_copy].copy_from_slice(&self.read_buffer[..to_copy]);
        self.read_buffer.drain(..to_copy);
        
        Ok(to_copy)
    }
    
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        // 简化实现：写入内部缓冲区
        self.write_buffer.extend_from_slice(buf);
        Ok(buf.len())
    }
    
    fn flush(&mut self) -> std::io::Result<()> {
        // 简化实现：清空写入缓冲区（模拟发送）
        self.write_buffer.clear();
        Ok(())
    }
    
    fn is_connected(&self) -> bool {
        self.connected
    }
    
    fn remote_addr(&self) -> Option<String> {
        self.remote_addr.clone()
    }
    
    fn shutdown(&mut self) -> std::io::Result<()> {
        self.connected = false;
        self.read_buffer.clear();
        self.write_buffer.clear();
        Ok(())
    }
    
    fn network_type(&self) -> &'static str {
        "Tokio-TCP"
    }
}

/// Tokio任务句柄适配器
pub struct TokioTaskHandle {
    handle: tokio::task::JoinHandle<()>,
    task_id: String,
}

impl TokioTaskHandle {
    pub fn new(handle: tokio::task::JoinHandle<()>, task_id: String) -> Self {
        Self { handle, task_id }
    }
}

impl TaskHandle for TokioTaskHandle {
    fn abort(&self) {
        self.handle.abort();
    }
    
    fn is_finished(&self) -> bool {
        self.handle.is_finished()
    }
    
    fn task_id(&self) -> String {
        self.task_id.clone()
    }
}

/// Tokio轻量级运行时适配器
/// 
/// 将Tokio的异步运行时适配为轻量级运行时接口
/// 使得协议层可以通过统一接口使用Tokio协程
pub struct TokioLightweightRuntime {
    /// 任务计数器
    task_counter: AtomicU64,
    /// 统计信息
    stats: Arc<std::sync::Mutex<RuntimeStats>>,
}

impl TokioLightweightRuntime {
    /// 创建新的Tokio轻量级运行时适配器
    pub fn new() -> Self {
        Self {
            task_counter: AtomicU64::new(0),
            stats: Arc::new(std::sync::Mutex::new(
                RuntimeStats::new("Tokio-Lightweight".to_string())
            )),
        }
    }
}

impl Default for TokioLightweightRuntime {
    fn default() -> Self {
        Self::new()
    }
}

impl LightweightRuntime for TokioLightweightRuntime {
    fn spawn_task<F>(&self, task: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let task_id = self.task_counter.fetch_add(1, Ordering::SeqCst);
        
        // 使用Tokio spawn_blocking来运行同步任务
        tokio::task::spawn_blocking(move || {
            task();
        });
        
        // 更新统计
        if let Ok(mut stats) = self.stats.lock() {
            stats.total_spawned += 1;
            stats.active_tasks += 1;  // 简化统计
        }
        
        println!("🚀 Tokio轻量级运行时派发任务 {}", task_id);
    }
    
    fn lightweight_type(&self) -> &'static str {
        "Tokio-Lightweight"
    }
}

/// Tokio完整运行时适配器
/// 
/// 实现完整的RuntimeAbstraction trait
pub struct TokioRuntimeAdapter {
    /// 任务计数器
    task_counter: AtomicU64,
    /// 统计信息
    stats: Arc<std::sync::Mutex<RuntimeStats>>,
}

impl TokioRuntimeAdapter {
    pub fn new() -> Self {
        Self {
            task_counter: AtomicU64::new(0),
            stats: Arc::new(std::sync::Mutex::new(
                RuntimeStats::new("Tokio-Full".to_string())
            )),
        }
    }
}

impl Default for TokioRuntimeAdapter {
    fn default() -> Self {
        Self::new()
    }
}

impl RuntimeAbstraction for TokioRuntimeAdapter {
    type Handle = TokioTaskHandle;
    
    fn spawn<F>(&self, future: F) -> Self::Handle
    where
        F: Future<Output = ()> + Send + 'static,
    {
        let task_id = format!("tokio-task-{}", 
                             self.task_counter.fetch_add(1, Ordering::SeqCst));
        
        let handle = tokio::spawn(future);
        
        // 更新统计
        if let Ok(mut stats) = self.stats.lock() {
            stats.total_spawned += 1;
            stats.active_tasks += 1;
        }
        
        TokioTaskHandle::new(handle, task_id)
    }
    
    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>> {
        Box::pin(tokio::time::sleep(duration))
    }
    
    fn timeout<F, T>(&self, duration: Duration, future: F) -> Pin<Box<dyn Future<Output = Result<T, crate::core::network::TimeoutError>> + Send>>
    where
        F: Future<Output = T> + Send + 'static,
        T: Send + 'static,
    {
        Box::pin(async move {
            match tokio::time::timeout(duration, future).await {
                Ok(result) => Ok(result),
                Err(_) => Err(crate::core::network::TimeoutError),
            }
        })
    }
    
    fn runtime_type(&self) -> &'static str {
        "Tokio-Full"
    }
    
    fn stats(&self) -> RuntimeStats {
        self.stats.lock().unwrap().clone()
    }
}

/// QUIC网络流适配器（模拟实现）
/// 
/// 展示如何为其他协议实现NetworkStream trait
#[derive(Debug)]
pub struct QuicNetworkStream {
    remote_addr: Option<String>,
    connected: bool,
    buffer: Vec<u8>,
}

impl QuicNetworkStream {
    pub fn new(remote_addr: String) -> Self {
        Self {
            remote_addr: Some(remote_addr),
            connected: true,
            buffer: Vec::new(),
        }
    }
}

impl NetworkStream for QuicNetworkStream {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        // QUIC特定的读取逻辑（模拟）
        if self.buffer.is_empty() {
            return Ok(0);
        }
        
        let to_copy = std::cmp::min(buf.len(), self.buffer.len());
        buf[..to_copy].copy_from_slice(&self.buffer[..to_copy]);
        self.buffer.drain(..to_copy);
        
        Ok(to_copy)
    }
    
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        // QUIC特定的写入逻辑（模拟）
        println!("📡 QUIC发送数据: {} bytes", buf.len());
        Ok(buf.len())
    }
    
    fn flush(&mut self) -> std::io::Result<()> {
        // QUIC特定的刷新逻辑
        Ok(())
    }
    
    fn is_connected(&self) -> bool {
        self.connected
    }
    
    fn remote_addr(&self) -> Option<String> {
        self.remote_addr.clone()
    }
    
    fn shutdown(&mut self) -> std::io::Result<()> {
        self.connected = false;
        println!("🔒 QUIC连接已关闭");
        Ok(())
    }
    
    fn network_type(&self) -> &'static str {
        "QUIC"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_tokio_network_stream() {
        let mut stream = TokioNetworkStream::mock_for_testing("127.0.0.1:6379".to_string());
        
        assert!(stream.is_connected());
        assert_eq!(stream.network_type(), "Tokio-TCP");
        assert_eq!(stream.remote_addr(), Some("127.0.0.1:6379".to_string()));
        
        // 测试写入和读取
        let test_data = b"hello world";
        assert_eq!(stream.write(test_data).unwrap(), test_data.len());
        
        println!("✅ Tokio网络流适配器测试通过");
    }
    
    #[test]
    fn test_tokio_lightweight_runtime() {
        let runtime = TokioLightweightRuntime::new();
        assert_eq!(runtime.lightweight_type(), "Tokio-Lightweight");
        
        // 注意：在测试环境中，tokio::spawn_blocking可能无法正常工作
        // 这里只测试类型和基本创建
        println!("✅ Tokio轻量级运行时适配器测试通过");
    }
    
    #[test]
    fn test_quic_network_stream() {
        let mut stream = QuicNetworkStream::new("127.0.0.1:443".to_string());
        
        assert!(stream.is_connected());
        assert_eq!(stream.network_type(), "QUIC");
        assert_eq!(stream.remote_addr(), Some("127.0.0.1:443".to_string()));
        
        // 测试QUIC特定功能
        let test_data = b"QUIC data";
        assert_eq!(stream.write(test_data).unwrap(), test_data.len());
        
        println!("✅ QUIC网络流适配器测试通过");
    }
    
    #[tokio::test]
    async fn test_async_creation() {
        // 测试从实际Tokio流创建适配器的过程
        // 由于测试环境限制，这里使用模拟的方式
        
        let stream = TokioNetworkStream::mock_for_testing("test:1234".to_string());
        assert_eq!(stream.network_type(), "Tokio-TCP");
        
        println!("✅ 异步创建测试通过");
    }
}